package com.bw.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.MyKafkaUtil;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.Properties;

public class Test3 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置并行度为1
        env.setParallelism(1);

        DataStreamSource<String> stream = env.addSource(MyKafkaUtil.getKafkaConsumer("tms_ods_yk9","test4"));

        //筛选订单数据
        SingleOutputStreamOperator<String> orderInfoDS = stream.filter(new FilterFunction<String>() {
            @Override
            public boolean filter(String s) throws Exception {
                JSONObject jsonObject = JSON.parseObject(s);
                String tableName = jsonObject.getString("table");
                if ("order_info".equals(tableName)) {
                    return true;
                }
                return false;
            }
        });

        orderInfoDS.print();
        //对流中的数据类型进行转换
        SingleOutputStreamOperator<JSONObject> mapDS = orderInfoDS.map(x -> JSON.parseObject(x));

        // 创建Tag

        //60020
        OutputTag<String> t60020 = new OutputTag<String>("t60020"){};
        //60030
        OutputTag<String> t60030 = new OutputTag<String>("t60030"){};
        //60050
        OutputTag<String> t60050 = new OutputTag<String>("t60050"){};
        //60060
        OutputTag<String> t60060 = new OutputTag<String>("t60060"){};
        //60070
        OutputTag<String> t60070 = new OutputTag<String>("t60070"){};
        //60080
        OutputTag<String> t60080 = new OutputTag<String>("t60080"){};
        //60999
        OutputTag<String> t60999 = new OutputTag<String>("t60999"){};

        SingleOutputStreamOperator<JSONObject> splitStream = mapDS.process(new ProcessFunction<JSONObject, JSONObject>() {
            @Override
            public void processElement(JSONObject jsonObject, ProcessFunction<JSONObject, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {
                String status = jsonObject.getJSONObject("data").getString("status");
                 if ("60020".equals(status)) {
                    context.output(t60020, jsonObject.toJSONString());
                } else if ("60030".equals(status)) {
                    context.output(t60030, jsonObject.toJSONString());
                } else if ("60050".equals(status)) {
                    context.output(t60050, jsonObject.toJSONString());
                } else if ("60060".equals(status)) {
                    context.output(t60060, jsonObject.toJSONString());
                } else if ("60070".equals(status)) {
                    context.output(t60070, jsonObject.toJSONString());
                } else if ("60080".equals(status)) {
                    context.output(t60080, jsonObject.toJSONString());
                } else if ("60999".equals(status)) {
                    context.output(t60999, jsonObject.toJSONString());
                }
            }
        });

        splitStream.map(new MapFunction<JSONObject, String>() {
            @Override
            public String map(JSONObject jsonObject) throws Exception {
                return jsonObject.toJSONString();
            }
        }).addSink(MyKafkaUtil.getKafkaProducer("dwd-order-detail_yk9"));

        splitStream.getSideOutput(t60020).addSink(MyKafkaUtil.getKafkaProducer("dwd-pay-detail_yk9"));
        splitStream.getSideOutput(t60030).addSink(MyKafkaUtil.getKafkaProducer("dwd-receive-detail_yk9"));
        splitStream.getSideOutput(t60050).addSink(MyKafkaUtil.getKafkaProducer("dwd-dispatch-detail_yk9"));
        splitStream.getSideOutput(t60060).addSink(MyKafkaUtil.getKafkaProducer("dwd-bound-detail_yk9"));
        splitStream.getSideOutput(t60070).addSink(MyKafkaUtil.getKafkaProducer("dwd-deliver-detail_yk9"));
        splitStream.getSideOutput(t60080).addSink(MyKafkaUtil.getKafkaProducer("dwd-sign-detail_yk9"));
        splitStream.getSideOutput(t60999).addSink(MyKafkaUtil.getKafkaProducer("dwd-cancel-detail_yk9"));


        env.execute();

    }
}
